<?php

/** 发送消息给gomq队列
 * 
 * @author chongi
 *
 * @method sign() 签名
 * 
 * 
 * 
 * @method push() 发送消息
 * 
 */
class messageHandler {
	
	
	
	public $scope='xshop';
	
	public $proxyUrl='10.128.7.234:8111/app';
	
	public $syncUrl='10.128.7.234:8111/app';
	
	public $callbackUrl='lppz.xxxxx.com/callback';
	
	private $isSign=False;
	
	
	/**
	 * 数据签名
	 * @param array $params	数据参数
	 * @param string $secertKey	私钥
	 * @return array 投递的消息数据
	 */
	public function sign($params,$secertKey='187a5c1ae5501d1e7a1df907365a66a9'){
	
		
		$unsigned_string = '';
		$request=array(
				'api'	=> 'a4a6cb8b60695d718a902afaba4c2765',
				
				'ver'	=> '1.0',
				'param'	=> json_encode($params),
				'req'	=> '',
		);
		unset($request['sign'], $request['scope']);
		ksort($request);
		
		
		foreach ($request as $key => $val) {
			$unsigned_string .= $key . $val;
		}
		$serverSign=md5($unsigned_string . $secertKey);
		$request['sign']=$serverSign;
		$request['scope']=$this->scope;
		
	
		$this->isSign=True;
		return array('action' => $this->scope,'operFlg' => 1,'params' => $request,'proxyUrl' => $this->proxyUrl,'callbackUrl' => $this->callbackUrl,'syncUrl' => $this->syncUrl);
	}
	
	/**
	 * 投递消息到后端应用队列MQ中
	 *
	 * @param any $message	投递的消息
	 * 						$message=array('action' => 'bill','operFlg' => 1,'params' => array('1201306170000000138'));
	 $message=array('action' => 'package','operFlg' => 1,'params' => array('1201306170000000138'));
	
	 action  string 任务 		: bill 订单
	 : package 包裹
	  
	 openFlg  int  操作标示  	: 1   新建
	 2  修改
	 3 状态变更
	
	 10 取消
	 11 异常
	
	 params	array 任务参数
	 	
	 * @param int 	$queueId  投递的队列ID MQ_QUEUE_BILL: 订单队列  :1 包裹队列  : 2
	 * @param int	$userId	  投递消息的用户标示,默认为0
	 * @param string $format 投递的消息封装格式,默认为json格式
	 *
	 * @return array
	 * array(
	 * 		'code' => int,
	 * 		'message' => string
	 * )
	 * 	code :				0 : 成功 1 : 消息协议错误  2 : 队列不存在  4 : 未知错误  5: socket 创建失败
	 * 						6 : 连接队列服务失败  	 7 : 消息发送长度异常                  8:  连接超时
	 * 						9 : 设置超时时间失败	10 : socket 写入失败		11  读取消息异常
	 * 						400 : 参数不存在  		 401 :未知的数据类型
	 */
	public function push($message,$isSync=False,$queueId=11,$userId='2e9be08ac83ac0725749e9e7079a475b',$format='json'){
	
		//检查参数
		if (! isset($message) or empty($message)){
			return array('code' => 400,'message' => '消息不存在');
		}
	
		if ( ! is_int($queueId) or $queueId <=0  ){
			return array('code' => 400,'message' => '队列ID类型错误');
		}
	
		$msg='';
		switch($format){
			case 'json' :
					
				if (is_string($message)){
					$msg=json_encode(array(strval($message)));
				}
				elseif (is_array($message)){
					$msg=json_encode($message);
				}
				elseif (is_object($message)){
					if (get_class($message) == 'entity'){
						$msg=json_encode($message->getFieldsMap());
					}
					else {
						$msg=json_encode($message);
					}
				}
				else {
					return array('code' => 401 ,'message' => '未知的message数据类型');
				}
				break;
			default :
				return array('code' => 401,'message' => '未知的消息封装格式');
					
		}
	
		//创建socket
		$socket=socket_create(AF_INET,SOCK_STREAM,SOL_TCP);
	
		if ($socket == FALSE){
			return array('code' => 5,'message' => 'socket创建失败');
		}
	
		//连接队列服务
	
		$connect=socket_connect($socket,'121.196.141.240',30004);
		if ($connect == FALSE ){
			return array('code' =>400,'message' => '连接队列服务失败');
		}
	
		socket_set_option($socket,SOL_SOCKET,SO_SNDBUF,1677721);
		socket_set_option($socket,SOL_SOCKET,SO_RCVBUF,1677721);
		if ($isSync){
			//设置socket超时
			$set_timeout = socket_set_option($socket,SOL_SOCKET,SO_RCVTIMEO,array('sec'	=> 5,'usec'	=> 5000	));
		}
		else {
			//设置socket超时
			$set_timeout = socket_set_option($socket,SOL_SOCKET,SO_RCVTIMEO,array('sec'	=> 1000,'usec'	=> 100000	));
		}
		if ($set_timeout == FALSE){
			socket_close($socket);
			return array('code' => 400,'message' => '设置连接超时失败');
		}
	
		/*组装消息体
		 message [
		客户端身份ID		BYTE	投递客户端默认ID为1
		队列ID			SHORT
		消息体 			STRING	以\0结尾
		]
		*/
	
	
		if ($isSync == True){
			$message_identity=3;
		}
		else {
			$message_identity=1;
		}
		$send_msg_len = strlen($msg) + 1;
		//$send_msg=pack('C' ,$message_identity) . pack('S',11) .pack('a32','a4084d6746132af4f5b26aa3e771f0b5') .  pack('a' . $send_msg_len,$msg) ;
		$send_msg=pack('C' ,$message_identity) . pack('S',$queueId) .pack('a32',$userId) .  pack('a' . $send_msg_len,$msg) ;
		
		$msg_len=strlen($send_msg);
	
		$i=0;
		$send_status=FALSE;
		$send_byte=0;
		$reset_count=3;
	
		//将二进制数据写入到socket中，如果写入字节数小于发送的长度，则默认允许3次重新写入剩余长度
		while($i < $reset_count){
	
			$send_byte+=socket_write($socket,$send_msg);
			if ($send_byte == FALSE){
				socket_close($socket);
				return array('code' => 400,'message' => 'socket 写入失败');
			}
			if ($send_byte < $msg_len){
				$send_msg=substr($send_msg,$send_byte);
			}
			if ($send_byte == $msg_len){
				$send_status = TRUE;
				break;
			}
			$i++;
		}
	
		if ($send_status === FALSE){
			socket_close($socket);
			return array('code' => 400,'message' => '消息长度发送异常');
		}
	
	
		//读取消息投递状态,返回1BYTE 状态标识
		$msg_status='';
		$msg_read_len=socket_recv($socket,$msg_status,1,0);
		if ($msg_read_len != 1 ){
			socket_close($socket);
			return array('code' => 400,'message' => '读取消息状态异常');
		}
		$msg_send_status_ary=unpack('C', $msg_status);
	
		// 0 : 投递成功  1 : 消息协议错误 2 : 队列不存在 3 : 未知错误
		// 0 : 投递成功  1 : 消息协议错误 2 : 队列不存在 3 : 未知错误
		
		switch($msg_send_status_ary[1]){
			case 0 :
				if ($isSync ==True){

					$relay_msg_length='';
					$msg_read_len=socket_recv($socket,$relay_msg_length,4,0);
					if ($msg_read_len != 4 ){
						socket_close($socket);
						return array('code' => 400,'message' => '读取消息内容异常');
					}
					$message_data_length=unpack('i', $relay_msg_length);
					
					$message_length=$message_data_length[1];
					
					$read_message_length=0;
					$read_message='';
					$i=0;
					while (($message_length-$read_message_length) != 0){
						if ($i >= 10000){
							return array('code' => 503 ,'message' => '尝试3次之后，仍未读取到回复内容');
						}
						$read_relay_message='';
						$msg_read_len=socket_recv($socket,$read_relay_message,$message_length,0);
						$read_message .= $read_relay_message;
						$read_message_length += strlen($read_relay_message);
						$i++;
					}
					
					$read_data=unpack('a*',$read_message);
					socket_close($socket);
					$returnResult=json_decode($read_data[1],true);
					if ($returnResult) {
						return $returnResult;
					}
					else {
						return array('code' => 200,'message' => '调用成功,但数据解码失败','data' => $read_data[1]);
					}
					
						
				}
		
				socket_close($socket);
				return array('code' => 200,'message' => '消息投递成功');
			case 1 :
				socket_close($socket);
				return array('code' => 400 ,'message' => '消息协议错误');
			case 2 :
				socket_close($socket);
				return array('code' => 404 ,'message' => '队列不存在');
			case 3 :
				socket_close($socket);
				return array('code' => 500 ,'message' => '未知错误');
			
		}
	}
}
